package cn.callmee.springboot.pulsar.starter.client.config;

import cn.callmee.springboot.pulsar.starter.client.PulsarProducerTemplate;
import cn.callmee.springboot.pulsar.starter.client.config.aspect.PulsarProducerAspect;
import cn.callmee.springboot.pulsar.starter.client.exceptions.ClientInitException;
import cn.callmee.springboot.pulsar.starter.client.properties.PulsarConsumerProperties;
import cn.callmee.springboot.pulsar.starter.client.properties.PulsarOauth2Properties;
import cn.callmee.springboot.pulsar.starter.client.properties.PulsarProperties;
import cn.callmee.springboot.pulsar.starter.client.properties.PulsarTlsProperties;
import com.google.common.base.Strings;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.client.impl.auth.oauth2.AuthenticationFactoryOAuth2;
import org.springframework.beans.BeansException;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.core.Ordered;
import org.springframework.lang.NonNull;

import javax.annotation.Resource;
import java.net.MalformedURLException;
import java.net.URL;
import java.util.concurrent.TimeUnit;

@Configuration
@ConditionalOnProperty(prefix = "pulsar", name = "enable", havingValue = "true")
@EnableConfigurationProperties({PulsarOauth2Properties.class, PulsarTlsProperties.class, PulsarConsumerProperties.class, PulsarProperties.class})
@Import({PulsarClientInitialConfiguration.class, PulsarProducerAspect.class, PulsarProducerTemplate.class})
public class PulsarStarterInitialConfiguration implements ApplicationContextAware, Ordered {
    public static org.springframework.context.ApplicationContext ApplicationContext;

    @Override
    public void setApplicationContext(@NonNull org.springframework.context.ApplicationContext applicationContext) throws
            BeansException {
        PulsarStarterInitialConfiguration.ApplicationContext = applicationContext;
    }

    @Override
    public int getOrder() {
        return Ordered.HIGHEST_PRECEDENCE;
    }

    // 自动配置
    @Resource
    private PulsarProperties pulsarProperties;

    @Bean
    @ConditionalOnMissingBean
    public PulsarClient pulsarClient() throws PulsarClientException, ClientInitException, MalformedURLException {
        PulsarTlsProperties pulsarTlsProperties = pulsarProperties.getTls();
        PulsarOauth2Properties pulsarOauth2Properties = pulsarProperties.getOauth2();

        if (!Strings.isNullOrEmpty(pulsarTlsProperties.getAuthCertFilePath()) &&
                !Strings.isNullOrEmpty(pulsarTlsProperties.getAuthKeyFilePath()) &&
                !Strings.isNullOrEmpty(pulsarProperties.getTokenAuthValue())
        ) throw new ClientInitException("You cannot use multiple auth options.");

        final ClientBuilder pulsarClientBuilder = PulsarClient.builder()
                .serviceUrl(pulsarProperties.getServiceUrl())
                .ioThreads(pulsarProperties.getIoThreads())
                .listenerThreads(pulsarProperties.getListenerThreads())
                .enableTcpNoDelay(pulsarProperties.isEnableTcpNoDelay())
                .keepAliveInterval(pulsarProperties.getKeepAliveIntervalSec(), TimeUnit.SECONDS)
                .connectionTimeout(pulsarProperties.getConnectionTimeoutSec(), TimeUnit.SECONDS)
                .operationTimeout(pulsarProperties.getOperationTimeoutSec(), TimeUnit.SECONDS)
                .startingBackoffInterval(pulsarProperties.getStartingBackoffIntervalMs(), TimeUnit.MILLISECONDS)
                .maxBackoffInterval(pulsarProperties.getMaxBackoffIntervalSec(), TimeUnit.SECONDS)
                .useKeyStoreTls(pulsarTlsProperties.isTlsUseKeyStore())
                .tlsTrustCertsFilePath(pulsarTlsProperties.getTrustCertsFilePath())
                .tlsCiphers(pulsarTlsProperties.getCiphers())
                .tlsProtocols(pulsarTlsProperties.getProtocols())
                .tlsTrustStorePassword(pulsarTlsProperties.getTrustStorePassword())
                .tlsTrustStorePath(pulsarTlsProperties.getTrustStorePath())
                .tlsTrustStoreType(pulsarTlsProperties.getTrustStoreType())
                .allowTlsInsecureConnection(pulsarTlsProperties.isAllowInsecureConnection())
                .enableTlsHostnameVerification(pulsarTlsProperties.isEnableHostnameVerification());

        if (!Strings.isNullOrEmpty(pulsarTlsProperties.getAuthCertFilePath()) &&
                !Strings.isNullOrEmpty(pulsarTlsProperties.getAuthKeyFilePath())) {
            pulsarClientBuilder.authentication(AuthenticationFactory
                    .TLS(pulsarTlsProperties.getAuthCertFilePath(), pulsarTlsProperties.getAuthKeyFilePath()));
        }

        if (!Strings.isNullOrEmpty(pulsarProperties.getTokenAuthValue())) {
            pulsarClientBuilder.authentication(AuthenticationFactory
                    .token(pulsarProperties.getTokenAuthValue()));
        }
        if (!Strings.isNullOrEmpty(pulsarOauth2Properties.getAudience()) &&
                !Strings.isNullOrEmpty(pulsarOauth2Properties.getIssuerUrl()) &&
                !Strings.isNullOrEmpty(pulsarOauth2Properties.getCredentialsUrl())) {
            final URL issuerUrl = new URL(pulsarOauth2Properties.getIssuerUrl());
            final URL credentialsUrl = new URL(pulsarOauth2Properties.getCredentialsUrl());

            pulsarClientBuilder.authentication(
                    AuthenticationFactoryOAuth2
                            .clientCredentials(issuerUrl, credentialsUrl, pulsarOauth2Properties.getAudience())
            );
        }
        if (!Strings.isNullOrEmpty(pulsarProperties.getListenerName())) {
            pulsarClientBuilder.listenerName(pulsarProperties.getListenerName());
        }

        return pulsarClientBuilder.build();
    }
}
